package cn.kgc.gmall.utils;

import cn.kgc.gmall.app.func.TransientSink;
import cn.kgc.gmall.bean.VisitorStats;
import cn.kgc.gmall.common.GmallConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import scala.annotation.meta.field;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * 连接clickHouse数据库往数据库中写数据
 */
public class ClickhouseUtil {

    public static <T> SinkFunction<T> getJdbcSink(String sql){
        // flink中的addSink需要sinkFunction
       return JdbcSink.sink( // flink的jdbc驱动包jdbcSink使用jdbc连接数据库之后返回sinkFunction
                // sql语句 insert into visitor_stats_2022(字段名...) values(?...)
                sql,
                // 为sql语句中的?进行赋值
                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement ps, T t) throws SQLException {
                        // 传进来的对象字段数必须与传进来的sql语句插入到字段数一致
                        // 一次的将对象中的属性值拿出来 赋值给?进行赋值
                        // 通过反射获取到对象中的属性
                        Field[] fields = t.getClass().getDeclaredFields();
                        // 即能够获取对象中的属性 也可以获取其父类中的属性
                        // Field[] fields = t.getClass().getFields();
                        int jumpIndex = 0;
                        for (int i = 0; i < fields.length;i++) {
                            Field field = fields[i];
                            // 如果对象的属性是私有的 设置为true也能访问
                            field.setAccessible(true);
                            // 判断字段有没有被打上注解  如果字段被打上了注解就跳过这个字段
                            // 通过字段获取注解对象
                            TransientSink annotation = field.getAnnotation(TransientSink.class);
                            // 如果字段上没有打上注解代表那么对象就是空的
                            if (annotation != null){
                                jumpIndex ++;
                                continue;
                            }
                            // 获取字段中的值
                            try {
                                // 所有的jdbc操作下标都是从1开始
                                ps.setObject(i+1 - jumpIndex,field.get(t));
                            } catch (IllegalAccessException e) {
                                throw new RuntimeException(e);
                            }

                        }

                    }
                },
                // 设置执行时的一些参数
                new JdbcExecutionOptions.Builder()
                        .withBatchSize(5) // 每5条数据为一批 去clickHouse中执行一次
                        //.withMaxRetries(3) // 最大重试次数 默认值3
                        //.withBatchIntervalMs(5000)// 如果5秒内没有5条数据也执行一批
                        .build(),
                // 连接数据库
                new JdbcConnectionOptions
                        .JdbcConnectionOptionsBuilder()
                        // 连接驱动
                        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                        // 连接地址
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .build()
        );
    }
}
